/*
 * $AIST_Release: 0.9.0 $
 * Copyright 2011 Information Technology Research Institute, National
 * Institute of Advanced Industrial Science and Technology
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 *    http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package aggr

import Aggregators._
import scala.collection.mutable.{Map => MMap}
import scala.collection.JavaConversions._

import com.google.protobuf.Descriptors._
import com.google.protobuf.Descriptors.FieldDescriptor.Type
import com.google.protobuf.Message
import com.google.protobuf.DynamicMessage

import util.JavaTypes._
import SerializedTypes._
import Util._

/**
 * "sum" aggregator
 */
class Sum(valType: SerializedType) extends Aggregator with Combiner {
  case class Tuple(content: MMap[FieldDescriptor, Any])

  def emptyTuple: Tuple = Tuple(MMap())

  override def combine(data: TraversableOnce[(Any, Any)])(writer: (Any, Any) => Unit): Unit = aggregate(data)(writer)

  override def aggregate(data: TraversableOnce[(Any, Any)])(writer: (Any, Any) => Unit): Unit = {

    var m: MMap[Any, Any] = MMap()

    data.foreach { case (k, v) =>
      (m.get(k), v) match {
        case (None,             r: Message) => m(k) = message2tuple(r)
        case (None,             r)          => m(k) = r
        case (Some(l: Double),  r: Double)  => m(k) = l + r
        case (Some(l: Float),   r: Float)   => m(k) = l + r
        case (Some(l: Int),     r: Int)     => m(k) = l + r
        case (Some(l: Long),    r: Long)    => m(k) = l + r
        case (Some(l: Tuple),   r: Message) => add(l, r)
        case _                              => error("Unexpect data type")
      }
    }

    m.foreach { case (k, v) => 
      v match {
        case x: Tuple => writer(k, tuple2message(valType.asInstanceOf[MessageType].descriptor, x))
        case x        => writer(k, v)
      }
    }
  }

  def tuple2message(d: Descriptor, t: Tuple): Message = {
    var b = DynamicMessage.newBuilder(d)

    for ((k, v) <- t.content) {
      v match {
        case v: Tuple => b.setField(k, tuple2message(k.getMessageType(), v))
        case v        => b.setField(k, v)
      }
    }
    b.build()
  }

  def message2tuple(m: Message): Tuple = {
    var t = emptyTuple

    for ((d, f) <- m.getAllFields()) {
      f match {
        case rv: JLong   => t.content(d) = rv.longValue
        case rv: JInt    => t.content(d) = rv.intValue
        case rv: JFloat  => t.content(d) = rv.floatValue
        case rv: JDouble => t.content(d) = rv.doubleValue
        case rv: Message => t.content(d) = message2tuple(rv)
        case _           => error("Unexpect data type")
      }
    }
    t
  }

  /**
   * This method executes "l += r".
   * Therefore this method writes new value to l argument.
   */
  def add(l: Tuple, r: Message): Unit = {
    for ((d, v) <- r.getAllFields()) {
      (l.content.get(d), v) match {
        case (None,             rv: Message) => l.content(d) = message2tuple(rv)
        case (None,             rv: JLong)   => l.content(d) = rv.longValue
        case (None,             rv: JInt)    => l.content(d) = rv.intValue
        case (None,             rv: JFloat)  => l.content(d) = rv.floatValue
        case (None,             rv: JDouble) => l.content(d) = rv.doubleValue
        case (Some(lv: Long),   rv: JLong)   => l.content(d) = lv + rv.longValue
        case (Some(lv: Int),    rv: JInt)    => l.content(d) = lv + rv.intValue
        case (Some(lv: Float),  rv: JFloat)  => l.content(d) = lv + rv.floatValue
        case (Some(lv: Double), rv: JDouble) => l.content(d) = lv + rv.doubleValue()
        case (Some(lv: Tuple),  rv: Message) => add(lv, rv)
        case _ => error("Unexpect data type.")
      }
    }
  }
}

class SumFactory extends AggregatorFactory {

    override def hasCombiner() = true

    override def createNew(
                  keyType: SerializedType,
                  valType: SerializedType,
                  args: List[Long]): Aggregator = {
      new Sum(valType)
    }

    override def check(keyType: SerializedType,
              valType: SerializedType,
              args: List[SerializedType], 
              options: Map[String, SerializedType]): Option[List[String]]= {

      Util.isNumber(valType) && args.isEmpty() && options.isEmpty()

      val aggrName = "sum"

      val argsErrors = checkNumOfArgs(aggrName, args.length, 0) 
      val optionErrors = checkOptionName(aggrName, options, List()) 
      val valErrors = if (!isNumber(valType)) List("Content type must be number.") else Nil

      (argsErrors ++ optionErrors ++ valErrors) match {
        case Nil => None
        case x   => Some(x)
      }
    }
}
